package com.mjf.spark.day11

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 有状态转化操作-Window
 *    窗口时长：计算内容的时间范围；
 *    滑动步长：隔多久触发一次计算。
 *    注意：这两者都必须为采集周期的整数倍。
 *
 */
object SparkStreaming03_Window {
  def main(args: Array[String]): Unit = {

    // 创建配置文件对象   注意：Streaming程序执行至少需要2个线程（1个用来采集，1一个用来计算），所以不能设置为local
    val conf: SparkConf = new SparkConf().setAppName("SparkStreaming").setMaster("local[*]")

    // 创建SparkStreaming程序执行入口对象   Seconds:时间间隔
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))

    // 从指定的端口获取数据
    val socketDS: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop103", 9999)

    // 设置窗口大小以及滑动步长   这两者都必须为采集周期的整数倍
    val windowDS: DStream[String] = socketDS.window(Seconds(6), Seconds(3))

    // 将DS转换为RDD进行操作    DS本身是没有提供排序函数的
    val resDS: DStream[(String, Int)] = windowDS
      .flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)

    resDS.print()

    // 启动采集器
    ssc.start()

    // 等待采集结束之后，终止程序
    ssc.awaitTermination()

  }
}
